#!/usr/bin/env python
# -*- encoding: utf-8 -*-

import pandas as pd

from xpy3lib.XRetryableQuery import XRetryableQuery
from xpy3lib.XRetryableSave import XRetryableSave
from sicost.AbstractDPJob import AbstractDPJob


class RDS2MPPJob(AbstractDPJob):
    """
    通过两个SQL去取来源表和目标表的表结构

    传入6个参数   来源标  和目标表  及开始时间结束时间

    将来源表那段时间的数据取出来  在存到目标表里
    """
    source_schema = None
    source_table = None
    target_schema = None
    target_table = None

    def __init__(self,
                 p_config=None,
                 p_db_conn_mpp=None,
                 p_db_conn_rds=None,
                 p_source_schema=None,
                 p_source_table=None,
                 p_target_schema=None,
                 p_target_table=None,
                 p_account_period_start=None,
                 p_account_period_end=None):

        super(RDS2MPPJob, self).__init__(p_config=p_config,
                                         p_db_conn_mpp=p_db_conn_mpp,
                                         p_db_conn_rds=p_db_conn_rds,
                                         p_db_conn_dbprod7=None,
                                         p_unit=None,
                                         p_account=None,
                                         p_cost_center=None,
                                         p_account_period_start=p_account_period_start,
                                         p_account_period_end=p_account_period_end)

        self.source_schema = p_source_schema
        self.source_table = p_source_table
        self.target_schema = p_target_schema
        self.target_table = p_target_table
        pass

    def do_execute(self):
        """
        """
        self.logger.info('RDS2MPPJob.do_execute')
        sql = " SELECT COLNAME FROM " \
              " SYSCAT.columns" \
              " WHERE 1=1 " \
              " AND TABSCHEMA  = '%s' AND TABNAME= '%s'" % (
                  self.source_schema, self.source_table)

        self.logger.info(sql)

        df1 = XRetryableQuery(p_db_conn=self.db_conn_rds, p_sql=sql, p_max_times=5).redo()
        success = df1.empty is False
        if success is False:
            return
        df1.columns = df1.columns.str.upper()
        self.logger.info('df1_properties-')
        # df1_properties = df1.columns.values
        df1_properties = df1['COLNAME'].values.tolist()
        self.logger.info(",".join(df1_properties))
        # account_period_start = self.account_period_start.astype(str)
        # account_period_end = self.account_period_end.astype(str)
        # account_period_start = self.account_period_start.strftime('%Y%m%tmp_dict%H%M%S')
        # account_period_end = self.account_period_end.strftime('%Y%m%tmp_dict%H%M%S')
        if 'PRODUCE_TIME' in df1_properties:
            time_field = 'PRODUCE_TIME'
            account_period_start = self.account_period_start
            account_period_end = self.account_period_end
        elif 'PRODUCE_DATE' in df1_properties:
            time_field = 'PRODUCE_DATE'
            account_period_start = self.account_period_start[0:8]
            account_period_end = self.account_period_end[0:8]
        elif 'DATE' in df1_properties:
            time_field = 'DATE'
            account_period_start = self.account_period_start[0:8]
            account_period_end = self.account_period_end[0:8]

        sql = " SELECT * FROM " \
              " %s.%s" \
              " WHERE 1=1 " \
              " AND %s >= '%s'" \
              " AND %s < '%s'" % (
                  self.source_schema, self.source_table, time_field, account_period_start, time_field,
                  account_period_end)

        self.logger.info(sql)

        df0 = XRetryableQuery(p_db_conn=self.db_conn_rds, p_sql=sql, p_max_times=5).redo()
        success = df0.empty is False
        if success is False:
            return
        df0.columns = df0.columns.str.upper()


        sql = " SELECT COLNAME FROM " \
              " SYSCAT.columns" \
              " WHERE 1=1 " \
              " AND TABSCHEMA  = '%s' AND TABNAME = '%s'" % (
                  self.target_schema, self.target_table)

        self.logger.info(sql)

        df2 = XRetryableQuery(p_db_conn=self.db_conn_mpp, p_sql=sql, p_max_times=5).redo()
        success = df2.empty is False
        if success is False:
            return
        df2.columns = df2.columns.str.upper()

        self.logger.info('df2_properties-')
        # df2_properties = df2.columns.values
        df2_properties = df2['COLNAME'].values.tolist()
        self.logger.info(",".join(df2_properties))
        if self.target_table != 'T_ADS_FACT_SICB_MO0101':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)
        if self.source_table == 'T_ADS_FACT_SICB_MO0101_DCL':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND UNIT_CODE in ('Q112','Q212','Q312','Q412','Q512')" \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)
        if self.source_table == 'T_ADS_FACT_SICB_MO0101_FCL':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND UNIT_CODE in ('Q314','Q414','Q214','Q114')" \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)
        if self.source_table == 'T_ADS_FACT_SICB_MO0101_RBAF':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND UNIT_CODE in ('Q503')" \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)
        if self.source_table == 'T_ADS_FACT_SICB_MO0101_RCM':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND UNIT_CODE in ('Q602','Q502','Q402','Q102','Q202','Q302' )" \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)
        if self.source_table == 'T_ADS_FACT_SICB_MO0101_SACL':
            sql = " DELETE FROM " \
                  " %s.%s" \
                  " WHERE 1=1 " \
                  " AND UNIT_CODE in ('Q418','Q518')" \
                  " AND %s >= '%s'" \
                  " AND %s < '%s'" % (
                      self.target_schema, self.target_table, time_field, account_period_start, time_field,
                      account_period_end)

            self.logger.info(sql)
            self.db_conn_mpp.execute(sql)


        self.logger.info('diff-1')
        diff1 = list(set(df1_properties).difference(set(df2_properties)))  # df1_properties中有而df2_properties中没有的
        self.logger.info(",".join(diff1))

        self.logger.info('diff-2')
        diff2 = list(set(df2_properties).difference(set(df1_properties)))  # df2_properties中有而df1_properties中没有的
        self.logger.info(",".join(diff2))

        # 删除多余列
        for p in diff1:
            df0.drop(p, axis=1, inplace=True)

        df0.drop(['REC_ID'], axis=1, inplace=True)

        XRetryableSave(p_db_conn=self.db_conn_mpp, p_table_name=self.target_table, p_schema=self.target_schema,
                       p_dataframe=df0,
                       p_max_times=5).redo()

        # self.__step_0()
        # self.__step_1()

        super(RDS2MPPJob, self).do_execute()
